package com.dongzili.common.disruptor.config;

import com.dongzili.common.disruptor.entity.MessageEvent;
import com.dongzili.common.disruptor.factory.MessageEventFactory;
import com.dongzili.common.disruptor.handler.MessageEventHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class MQManager {

    // 定义线程池配置
    @Bean("disruptorExecutor")
    public ThreadPoolExecutor disruptorExecutor() {
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        int maxPoolSize = corePoolSize * 2;
        int queueCapacity = 1024;

        return new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadFactory() {
                    private final AtomicInteger threadId = new AtomicInteger(1);
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "disruptor-thread-" + threadId.getAndIncrement());
                        thread.setDaemon(true);
                        return thread;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
    }

//    @Bean
//    public <T> MessageEventHandler<T> messageEventHandler() {
//        return new MessageEventHandler<>();
//    }

    @SuppressWarnings("deprecation")
    @Bean("MessageEvent")
    public <T> RingBuffer<MessageEvent<T>> messageVoRingBuffer(
            ThreadPoolExecutor disruptorExecutor,
            MessageEventHandler<T> messageEventHandler) {

        // 指定ringbuffer大小，必须为2的N次方
        int bufferSize = calculateRingBufferSize();

        // 创建Disruptor实例
        Disruptor<MessageEvent<T>> disruptor = new Disruptor<>(
                new MessageEventFactory<>(),
                bufferSize,
                disruptorExecutor,
                ProducerType.SINGLE,
                createWaitStrategy()
        );

        // 设置事件处理器
        disruptor.handleEventsWith(messageEventHandler);

        // 启动Disruptor
        Runtime.getRuntime().addShutdownHook(new Thread(disruptor::shutdown));
        disruptor.start();

        return disruptor.getRingBuffer();
    }

    // 计算最佳RingBuffer大小
    private int calculateRingBufferSize() {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        // 经验值：CPU核心数 * 131072 (128KB)
        return Math.max(1024 * 256, cpuCores * 131072);
    }

    // 根据场景选择等待策略
    private WaitStrategy createWaitStrategy() {
        // 低延迟场景使用忙等待策略
        if (isLowLatencyScenario()) {
            return new BusySpinWaitStrategy();
        }
        // 高吞吐量场景使用BlockingWaitStrategy
        return new BlockingWaitStrategy();
    }

    // 自定义场景判断逻辑
    private boolean isLowLatencyScenario() {
        // 可根据配置或系统负载动态判断
        return true; // 默认使用低延迟策略
    }
}
 